package ar.edu.itba.pod.legajo45129;

import java.io.Serializable;
import java.rmi.RemoteException;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

import ar.edu.itba.pod.legajo45129.simul.communication.ClusterAdministrationImpl;
import ar.edu.itba.pod.legajo45129.simul.communication.ConnectionManagerImpl;
import ar.edu.itba.pod.legajo45129.simul.communication.MessageListenerImpl;
import ar.edu.itba.pod.legajo45129.simul.communication.SimulationCommunicationImpl;
import ar.edu.itba.pod.legajo45129.simul.communication.factory.MessageFactory;
import ar.edu.itba.pod.legajo45129.simul.communication.payload.NodeAgentLoadPayloadImpl;
import ar.edu.itba.pod.legajo45129.statistics.Statistics;
import ar.edu.itba.pod.simul.communication.ConnectionManager;
import ar.edu.itba.pod.simul.communication.Message;
import ar.edu.itba.pod.simul.communication.NodeAgentLoad;
import ar.edu.itba.pod.simul.communication.payload.DisconnectPayload;
import ar.edu.itba.pod.simul.communication.payload.NodeAgentLoadPayload;
import ar.edu.itba.pod.simul.communication.payload.NodeMarketDataPayload;
import ar.edu.itba.pod.simul.communication.payload.ResourceRequestPayload;
import ar.edu.itba.pod.simul.communication.payload.ResourceTransferMessagePayload;
import ar.edu.itba.pod.simul.market.Resource;

import com.google.common.base.Preconditions;
import com.google.common.collect.Multiset;

public class MessageProcessor implements Runnable, Serializable {


	/**
	 * 
	 */
	private static final long serialVersionUID = 3859452774991807048L;

	private static Logger logger = Logger.getLogger(MessageProcessor.class);
	private boolean finish = false;

	private ConnectionManagerImpl connectionManager;
	private ConcurrentHashMap<String, Integer> agentLoads = new ConcurrentHashMap<String, Integer>();
//	private Processor messageProcessor = new Processor();

	public static MessageProcessor instance;

	public static MessageProcessor getInstance(ConnectionManager connectionManager) {
		if (instance == null)  {
			instance = new MessageProcessor(connectionManager);
		}

		return instance;
	} 


	private MessageProcessor(ConnectionManager connectionManager) {
		this.connectionManager = (ConnectionManagerImpl) connectionManager;
	}


	@Override
	public void run() {
		MessageListenerImpl messageListener = null;
		try {
			messageListener = (MessageListenerImpl) connectionManager.getGroupCommunication().getListener();
		} catch (RemoteException e) {
			logger.error("error listening for message");
			throw new IllegalStateException(e.getMessage());
		} 

		logger.info("message service is up");
		BlockingQueue<Message> queue = messageListener.getInbox();


		Message message = null;
		while (!finish) {
			if ((message = queue.peek()) != null) {
				logger.info(message +  " has arrived");
				new Thread(new Processor(message)).start();
				
				if (queue.remove(message)) {
					logger.debug("message removed from the inbox");
				} else {
					logger.debug("message coult not removed from the inbox");
				}
				logger.info("message processed");
			}
		}

	}

	/**
	 * clean all NODE_AGENTS_LOAD response received.
	 */
	public void resetLoads() {
		agentLoads.clear();
	}

	public NodeAgentLoad getMinimumNodeLoad() {
		Iterator<String> it = agentLoads.keySet().iterator();
		NodeAgentLoad ret = new NodeAgentLoad(connectionManager.getNodeId().toString(), Integer.MAX_VALUE);
		String nodeId;
		while (it.hasNext()) {
			nodeId = it.next();
			int load = agentLoads.get(nodeId);
			if (load < ret.getNumberOfAgents()) {
				ret = new NodeAgentLoad(nodeId, load);
			}
		}

		return ret;
	}

	private class Processor implements Runnable {
		private Message message;
		
		public Processor(Message message) {
			this.message = message;

		}
		
		public void process(Message message) throws RemoteException {
			switch(message.getType()) {
			case NODE_AGENTS_LOAD_REQUEST:
				processNodeAgentsLoadRequest(message);
				break;

			case NODE_AGENTS_LOAD:
				processNodeAgentLoad(message);
				break;

			case DISCONNECT:
				processDisconnect(message);
				break;

			case RESOURCE_TRANSFER:				// Resource been transfer from one node to another node
				processResourceTransfer(message);
				break;

			case RESOURCE_REQUEST:				// Resource request from one node to the cluster
				processResourceRequest(message);
				break;

			case NODE_MARKET_DATA:				// Node market data
				processNodeMarketData(message);
				break;
			case NODE_MARKET_DATA_REQUEST:		// Node market data request to all nodes
				processNodeMarketDataRequest(message);
				break;



				// not used!				
			case RESOURCE_TRANSFER_CANCELED:		// Resource transfer from one node to another node canceled
			case NEW_MESSAGE_REQUEST:			// New messages request from one node to another node
			case NEW_MESSAGE_RESPONSE:			// New messages response from one node to another node
				logger.debug("NOT USED");

			}
		}
		
		private void processNodeMarketDataRequest(Message message) {	
			Message response = MessageFactory.NodeMarketData(connectionManager.getNodeId().toString(), 
					connectionManager.getMarketManager().market().marketData());
			try {
				connectionManager.getGroupCommunication().send(response, message.getNodeId());
			} catch (RemoteException e) {
				logger.error("error sending the response to NODE_MARKET_DATA from: " + message.getNodeId());
			}			
		}

		private void processNodeMarketData(Message message) {
			
			NodeMarketDataPayload payload = (NodeMarketDataPayload) message.getPayload();
			Statistics.getInstance().addStatistics(message.getNodeId().toString(), payload.getMarketData().getHistory());
			Statistics.getInstance().printOnScreen();
		}

		private void processResourceRequest(Message message) {
			logger.debug("processing resoucers request- message");
			ResourceRequestPayload payload = (ResourceRequestPayload) message.getPayload();
			String localNode = connectionManager.getNodeId().toString();

			logger.debug("localNode: " + localNode);
			logger.debug("message from: " + message.getNodeId());

			if (localNode.equals(message.getNodeId())) {
				// do nothing if resourcesRequest come from local node
				logger.debug("message from the local node");
				return;
			}

			int stock = connectionManager.getMarketManager().market().getStock(payload.getResource(), payload.getAmountRequested());
			//			}

			logger.debug("new amount requested: " + stock);

			try{
				// begin the transaction
				logger.debug("begin transaction");
				connectionManager.getNodeCommunication().beginTransaction(
						message.getNodeId(), Configuration.TIMEOUT_TRANSACTION);

				// exchange my resources
				connectionManager.getNodeCommunication().exchange(payload.getResource(), 
						stock, localNode, 
						message.getNodeId());

				// end the transaction
				connectionManager.getNodeCommunication().endTransaction();
				logger.debug("transaction done");
			} catch (Exception e) {
				logger.error("error during the transaction: " + e.getMessage());

				try{
					// rollback the transactions
					logger.info("rollback the transaction");
					connectionManager.getNodeCommunication().rollback();
				} catch (Exception e1) {
					logger.error("error in rollback transaction: " + e1.getMessage());
				}
			}

		}

		private void processResourceTransfer(Message message) {
			logger.debug("processing resource transfer message");
			ResourceTransferMessagePayload payload = (ResourceTransferMessagePayload) message.getPayload();

			if(payload.getDestination().equals(connectionManager.getNodeId().toString())){
				logger.debug("adding resources to selling");
				Multiset<Resource> stock = connectionManager.getMarketManager().market().getIncomingResourcesStock();
				stock.setCount(payload.getResource(), stock.count(payload.getResource()) + payload.getAmount());
				logger.debug("resources added");

			} else {
				logger.debug("removing resources");
				connectionManager.getMarketManager().market().removeFromSelling(payload.getResource(), payload.getAmount());
				logger.debug("resources removed");
			}
		}

		private void processDisconnect(Message message) throws RemoteException {
			logger.debug("processing disconnect message");
			DisconnectPayload dp = (DisconnectPayload) message.getPayload();
			String node = dp.getDisconnectedNodeId();
			ClusterAdministrationImpl cai = (ClusterAdministrationImpl) connectionManager.getClusterAdmimnistration();
			if (connectionManager.getNodeId().toString().equals(node)) {
				// remove all my known node and disconnect from my group
				logger.debug("remove my known nodes");
				cai.disconnect();

			} else {
				// remove this node from my known nodes
				logger.debug("remove this node from my known nodes ");
				cai.removeNode(node);	
			}			
		}

		private void processNodeAgentLoad(Message message) {
			logger.debug("processing node agent load message");
			NodeAgentLoadPayload nodeAgentLoad = (NodeAgentLoadPayload) message.getPayload();
			agentLoads.put(message.getNodeId(), nodeAgentLoad.getLoad());
			logger.debug("node agent load message processed");
		}


		private void processNodeAgentsLoadRequest(Message message) throws RemoteException {
			logger.debug("processing node agents load request message");
			SimulationCommunicationImpl sci = (SimulationCommunicationImpl) connectionManager.getSimulationCommunication();
			NodeAgentLoadPayloadImpl payload = new NodeAgentLoadPayloadImpl(sci.getAgentsCount());
			Message response = MessageFactory.NodeAgentsLoad(connectionManager.getNodeId().toString(), payload);
			connectionManager.getGroupCommunication().send(response, message.getNodeId());
			logger.debug("node agents load message sent");
		}

		@Override
		public void run() {
			try {
				process(message);
			} catch (RemoteException e) {
				logger.error("could not process the message: " +  message);
			}
		}


	}

	/**
	 * Get all response of the message NODE_AGENTS_LOAD
	 * @return Map which key as node, and value as it node agent load
	 */

	public ConcurrentHashMap<String, Integer> getAgentsLoad() {
		Preconditions.checkNotNull(agentLoads);
		return agentLoads;
	}

	/**
	 *  finish this process
	 */
	public void shutDown() {
		finish= true;
	}

}
